package com.learning.netease.distributed.lock.zk.zklock.base;

import com.learning.netease.distributed.lock.zk.zklock.MyZkSerializer;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * ClassName: ZkDistributedLock
 * Description: 基于ZK实现的分布式锁
 * <p>
 * 基于临时节点获取实现，缺点：惊群效应，不建议使用。只有一个能抢到锁，一旦释放，其他的全部再去抢
 * <p>
 * Date: 2019/3/20 14:19 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class ZkDistributedLock implements Lock {

    private String lockPath;

    private ZkClient client;

    public ZkDistributedLock(String host, String lockPath) {
        super();
        this.lockPath = lockPath;
        this.client = new ZkClient(host);
        client.setZkSerializer(new MyZkSerializer());
    }

    /**
     * 尝试获取锁：创建临时节点（同一个ZNode下，节点的名称是唯一的）
     * 创建成功，获取锁
     * 创建实现（已经存在就会创建失败），没有获取锁
     *
     * @return
     */
    @Override
    public boolean tryLock() {
        try {
            // 创建临时节点，不能阻塞
            client.createEphemeral(lockPath);
        } catch (ZkNodeExistsException e) {
            return false;
        }
        return true;
    }

    @Override
    public void lock() {
        // 获取获取不到锁，则阻塞
        if (!tryLock()) {
            // 阻塞自己，持续获取锁
            waitForLock();
            // 再次尝试
            lock();
        }
    }

    /**
     * 注册watcher、阻塞自己
     */
    private void waitForLock() {
        CountDownLatch countDownLatch = new CountDownLatch(1);

        // 创建一个监听
        IZkDataListener listener = new IZkDataListener() {
            @Override
            public void handleDataChange(String dataPath, Object data) throws Exception {

            }

            @Override
            public void handleDataDeleted(String dataPath) throws Exception {
                System.out.println("***** 收到节点被删除了 *****");

                // 节点删除，意味着锁释放了，可以去获取
                countDownLatch.countDown();
            }
        };

        // 注册监听
        client.subscribeDataChanges(lockPath, listener);

        // 阻塞自己，如果节点已经存在则阻塞
        if (client.exists(lockPath)) {
            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        // 取消监听，方法退出，后续会继续调用lock方法
        client.unsubscribeDataChanges(lockPath, listener);

    }

    /**
     * 释放锁。删除节点
     */
    @Override
    public void unlock() {
        client.delete(lockPath);
    }


    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }
}
